| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318 |
2x
2x
2x
2x
2x
2x
425x
2x
2x
590x
2619x
590x
590x
590x
2x
431x
431x
431x
431x
425x
425x
425x
431x
431x
431x
431x
425x
425x
425x
6x
2x
275x
275x
275x
275x
274x
274x
274x
274x
275x
271x
271x
2x
2897x
1209x
1209x
1209x
1220x
1220x
1209x
2x
4x
4x
4x
4x
4x
2x
662x
662x
475x
475x
2x
2x
439x
439x
439x
439x
439x
2x
1251x
1251x
358x
358x
284x
270x
358x
1251x
842x
410x
409x
396x
1251x
2x
5x
2x
909x
909x
12x
2x
1154x
1154x
326x
828x
828x
382x
382x
446x
2x
409x
348x
61x
61x
60x
1x
2x
422x
422x
422x
422x
2x
422x
422x
477x
422x
2x
| /**
* Copyright 2017 Google Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { Query } from './query';
import { SyncEngine } from './sync_engine';
import { OnlineState, TargetId } from './types';
import { DocumentViewChange } from './view_snapshot';
import { ChangeType, ViewSnapshot } from './view_snapshot';
import { DocumentSet } from '../model/document_set';
import { assert } from '../util/assert';
import { EventHandler } from '../util/misc';
import { ObjectMap } from '../util/obj_map';
/**
* Holds the listeners and the last received ViewSnapshot for a query being
* tracked by EventManager.
*/
class QueryListenersInfo {
viewSnap: ViewSnapshot | null;
targetId: TargetId;
listeners: QueryListener[] = [];
}
/**
* Interface for handling events from the EventManager.
*/
export interface Observer<T> {
next: EventHandler<T>;
error: EventHandler<Error>;
}
/**
* EventManager is responsible for mapping queries to query event emitters.
* It handles "fan-out". -- Identical queries will re-use the same watch on the
* backend.
*/
export class EventManager {
private queries = new ObjectMap<Query, QueryListenersInfo>(q =>
q.canonicalId()
);
private onlineState: OnlineState = OnlineState.Unknown;
constructor(private syncEngine: SyncEngine) {
this.syncEngine.subscribe(
this.onChange.bind(this),
this.onError.bind(this)
);
}
listen(listener: QueryListener): Promise<TargetId> {
const query = listener.query;
let firstListen = false;
let queryInfo = this.queries.get(query);
if (!queryInfo) {
firstListen = true;
queryInfo = new QueryListenersInfo();
this.queries.set(query, queryInfo);
}
queryInfo.listeners.push(listener);
listener.applyOnlineStateChange(this.onlineState);
if (queryInfo.viewSnap) listener.onViewSnapshot(queryInfo.viewSnap);
if (firstListen) {
return this.syncEngine.listen(query).then(targetId => {
queryInfo!.targetId = targetId;
return targetId;
});
} else {
return Promise.resolve(queryInfo.targetId);
}
}
async unlisten(listener: QueryListener): Promise<void> {
const query = listener.query;
let lastListen = false;
const queryInfo = this.queries.get(query);
if (queryInfo) {
const i = queryInfo.listeners.indexOf(listener);
Eif (i >= 0) {
queryInfo.listeners.splice(i, 1);
lastListen = queryInfo.listeners.length === 0;
}
}
if (lastListen) {
this.queries.delete(query);
return this.syncEngine.unlisten(query);
}
}
onChange(viewSnaps: ViewSnapshot[]): void {
for (const viewSnap of viewSnaps) {
const query = viewSnap.query;
const queryInfo = this.queries.get(query);
Eif (queryInfo) {
for (const listener of queryInfo.listeners) {
listener.onViewSnapshot(viewSnap);
}
queryInfo.viewSnap = viewSnap;
}
}
}
onError(query: Query, error: Error): void {
const queryInfo = this.queries.get(query);
Eif (queryInfo) {
for (const listener of queryInfo.listeners) {
listener.onError(error);
}
}
// Remove all listeners. NOTE: We don't need to call syncEngine.unlisten()
// after an error.
this.queries.delete(query);
}
applyOnlineStateChange(onlineState: OnlineState): void {
this.onlineState = onlineState;
this.queries.forEach((_, queryInfo) => {
for (const listener of queryInfo.listeners) {
listener.applyOnlineStateChange(onlineState);
}
});
}
}
export interface ListenOptions {
/** Raise events when only metadata of documents changes */
readonly includeDocumentMetadataChanges?: boolean;
/** Raise events when only metadata of the query changes */
readonly includeQueryMetadataChanges?: boolean;
/**
* Wait for a sync with the server when online, but still raise events while
* offline.
*/
readonly waitForSyncWhenOnline?: boolean;
}
/**
* QueryListener takes a series of internal view snapshots and determines
* when to raise the event.
*
* It uses an Observer to dispatch events.
*/
export class QueryListener {
/**
* Initial snapshots (e.g. from cache) may not be propagated to the wrapped
* observer. This flag is set to true once we've actually raised an event.
*/
private raisedInitialEvent = false;
private options: ListenOptions;
private snap: ViewSnapshot;
private onlineState: OnlineState = OnlineState.Unknown;
constructor(
readonly query: Query,
private queryObserver: Observer<ViewSnapshot>,
options?: ListenOptions
) {
this.options = options || {};
}
onViewSnapshot(snap: ViewSnapshot): void {
assert(
snap.docChanges.length > 0 || snap.syncStateChanged,
'We got a new snapshot with no changes?'
);
if (!this.options.includeDocumentMetadataChanges) {
// Remove the metadata only changes.
const docChanges: DocumentViewChange[] = [];
for (const docChange of snap.docChanges) {
if (docChange.type !== ChangeType.Metadata) {
docChanges.push(docChange);
}
}
snap = new ViewSnapshot(
snap.query,
snap.docs,
snap.oldDocs,
docChanges,
snap.fromCache,
snap.hasPendingWrites,
snap.syncStateChanged
);
}
if (!this.raisedInitialEvent) {
if (this.shouldRaiseInitialEvent(snap, this.onlineState)) {
this.raiseInitialEvent(snap);
}
} else if (this.shouldRaiseEvent(snap)) {
this.queryObserver.next(snap);
}
this.snap = snap;
}
onError(error: Error): void {
this.queryObserver.error(error);
}
applyOnlineStateChange(onlineState: OnlineState): void {
this.onlineState = onlineState;
if (
this.snap &&
!this.raisedInitialEvent &&
this.shouldRaiseInitialEvent(this.snap, onlineState)
) {
this.raiseInitialEvent(this.snap);
}
}
private shouldRaiseInitialEvent(
snap: ViewSnapshot,
onlineState: OnlineState
): boolean {
assert(
!this.raisedInitialEvent,
'Determining whether to raise first event but already had first event'
);
// Always raise the first event when we're synced
if (!snap.fromCache) {
return true;
}
// NOTE: We consider OnlineState.Unknown as online (it should become Failed
// or Online if we wait long enough).
const maybeOnline = onlineState !== OnlineState.Failed;
// Don't raise the event if we're online, aren't synced yet (checked
// above) and are waiting for a sync.
if (this.options.waitForSyncWhenOnline && maybeOnline) {
assert(
snap.fromCache,
'Waiting for sync, but snapshot is not from cache'
);
return false;
}
// Raise data from cache if we have any documents or we are offline
return !snap.docs.isEmpty() || onlineState === OnlineState.Failed;
}
private shouldRaiseEvent(snap: ViewSnapshot): boolean {
// We don't need to handle includeDocumentMetadataChanges here because
// the Metadata only changes have already been stripped out if needed.
// At this point the only changes we will see are the ones we should
// propagate.
if (snap.docChanges.length > 0) {
return true;
}
const hasPendingWritesChanged =
this.snap && this.snap.hasPendingWrites !== snap.hasPendingWrites;
if (snap.syncStateChanged || hasPendingWritesChanged) {
return this.options.includeQueryMetadataChanges === true;
}
// Generally we should have hit one of the cases above, but it's possible
// to get here if there were only metadata docChanges and they got
// stripped out.
return false;
}
private raiseInitialEvent(snap: ViewSnapshot): void {
assert(
!this.raisedInitialEvent,
'Trying to raise initial events for second time'
);
snap = new ViewSnapshot(
snap.query,
snap.docs,
DocumentSet.emptySet(snap.docs),
QueryListener.getInitialViewChanges(snap),
snap.fromCache,
snap.hasPendingWrites,
true
);
this.raisedInitialEvent = true;
this.queryObserver.next(snap);
}
/** Returns changes as if all documents in the snap were added. */
private static getInitialViewChanges(
snap: ViewSnapshot
): DocumentViewChange[] {
const result: DocumentViewChange[] = [];
snap.docs.forEach(doc => {
result.push({ type: ChangeType.Added, doc });
});
return result;
}
}
|